package kafka

import (
	"AR/saas/toollib/logger"
	"fmt"
	"github.com/IBM/sarama"
	"time"
)

// ListenConsumer 消费者
func ListenConsumer() {
	consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
	if err != nil {
		logger.Error(fmt.Sprintf("new consumer err:%v", err))
		return
	}
	partitionList, err := consumer.Partitions("web_log")
	if err != nil {
		logger.Error(fmt.Sprintf("get partitions err:%v", err))
		return
	}
	go func() {
		for {
			// 遍历所有的分区:Topic
			for partition := range partitionList {
				pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
				if err != nil {
					fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
					return
				}
				go func(sarama.PartitionConsumer) {
					for msg := range pc.Messages() {
						// todo: 数据处理
						fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)

					}
				}(pc)
				err1 := pc.Close()
				if err1 != nil {
					logger.Error(fmt.Sprintf("Consume close err:%v", err1))
					return
				}
			}
			time.Sleep(time.Millisecond * 5)
		}
	}()
	for partition := range partitionList { // 遍历所有的分区
		// 针对每个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
			return
		}
		defer pc.AsyncClose()
		// 异步从每个分区消费信息
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
			}
		}(pc)
	}
}
